#include "common.hpp"

#include <thread>
#include <atomic>
#include <mutex>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <assert.h>

using namespace std;

class CommonSource
{
public:
	CommonSource(int total)
		:total_(total)
	{
		assert(total_ > 0);
	}

	virtual ~CommonSource() {}

	void Push(const string& msg)
	{
		unique_lock<mutex> guard(mtx_queue_);
		queue_.push_back(msg);
		push_cnt_.store(push_cnt_.load() + 1);
		cond_queue_.notify_all();
	}

	void Notify()
	{
		unique_lock<mutex> guard(mtx_queue_);
		cond_queue_.notify_all();
	}

	bool FinishConsume()
	{
		unique_lock<mutex> guard2(mtx_cnt_);
		return fetch_cnt_ == total_ && fetch_cnt_ > 0;
	}

	void Fetch(string& msg)
	{
		{
			unique_lock<mutex> guard(mtx_queue_);
			cond_queue_.wait(guard, [&]()->bool {
				return !queue_.empty(); });
			msg = queue_.front();
			queue_.pop_front();
			mtx_queue_.unlock();
		}
		{
			unique_lock<mutex> guard2(mtx_cnt_);
			fetch_cnt_++;
		}
	}

	int total_ = 0;
	atomic<int> push_cnt_;

	int fetch_cnt_ = 0;
	mutex mtx_cnt_;

	mutex mtx_queue_;
	deque<string> queue_;
	condition_variable cond_queue_;
};

class Worker
{
public:
	Worker(CommonSource* src, int id)
		:id_(id), src_(src)
	{
		thd_ = new thread(&Worker::Run, this);
	}

	virtual ~Worker()
	{
		if (thd_)
		{
			if (thd_->joinable())
			{
				thd_->join();
			}
			delete thd_;
			thd_ = NULL;
		}
	}

	void Join()
	{
		if (thd_ && thd_->joinable())
			thd_->join();
	}

	virtual void Run() = 0;

protected:
	int id_ = 0;

	CommonSource* src_ = NULL;
private:
	thread* thd_ = NULL;
};

class Producer :public Worker
{
public:
	Producer(CommonSource* src, int id, int produce_size)
		:Worker(src, id),
		produce_size_(produce_size)
	{

	}

	void Run()
	{
		for (int i = 0; i < produce_size; i++)
		{
			stringstream sss;
			sss << "p" << id_ << "_" << i;
			src_->Push(sss.str());
			this_thread::sleep_for(chrono::milliseconds(100));
		}
		printf("producer[%d] finish push\n", id_);
		while (!src_->FinishConsume())
		{
			src_->Notify();
			this_thread::sleep_for(chrono::milliseconds(100));
		}
		printf("producer[%d] finish\n", id_);
	}
private:
	int produce_size_ = 0;
};

class Consumer :public Worker
{
public:
	Consumer(CommonSource* src, int id)
		:Worker(src, id)
	{

	}

	void Run()
	{
		string msg;
		while (!src_->FinishConsume())
		{
			src_->Fetch(msg);
			printf("consume[%d] %s\n", id_, msg.c_str());
			this_thread::sleep_for(chrono::milliseconds(300));
		}
		printf("consume[%d] finish\n", id_);
	}
};

int main()
{
	CommonSource src(20);
	Producer pro1(&src, 1, 10);
	Producer pro2(&src, 2, 10);
	Consumer con1(&src, 1);
	Consumer con2(&src, 2);
	Consumer con3(&src, 3);
	pro1.Join();
	pro2.Join();
	con1.Join();
	con2.Join();
	con3.Join();
	return 1;
}